package com.atguigu.demo07.config;

import com.alibaba.fastjson.JSON;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
@Slf4j
@Configuration
public class MyKafkaConfig {
    @Resource

    //新创建主题：如果已存在直接复用
    @Bean
    public NewTopic myClusterTopic1(){
        return TopicBuilder
                .name("topic2") //主题名称
                .partitions(3) //分区数
                .replicas(3) // 副本数
                .build();

    }

//    @Bean
//    public NewTopic myClusterTopic3(){
//        return TopicBuilder
//                .name("my_cluster_topic3") //主题名称
//                .partitions(3) //分区数
//                .replicas(3) // 副本数
//                .build();
//
//    }
//    @Bean
//    public NewTopic myClusterTopic4(){
//        return TopicBuilder
//                .name("my_cluster_topic4") //主题名称
//                .partitions(3) //分区数
//                .replicas(3) // 副本数
//                .build();
//
//    }

//    @Bean
//    public NewTopic myClusterTopic5(){
//        return TopicBuilder
//                .name("my_cluster_topic5") //主题名称
//                .partitions(3) //分区数
//                .replicas(3) // 副本数
//                .build();
//
//    }

    //配置异常处理器
    @Bean
    public ConsumerAwareListenerErrorHandler errorHandler(){
        return new ConsumerAwareListenerErrorHandler(){

            @Override
            public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {
                //消费者消费出现异常时调用：一般结合手动ack使用
//                consumer
                log.error("消费异常：header={} , payload={}, exception={}" , JSON.toJSONString(message.getHeaders()) ,
                        JSON.toJSONString(message.getPayload()),
                        e.getMessage());
                //消息消费异常：一般使用重试机制 配置死信队列处理
                return null;
            }
        };
    }

//    @Bean
//    public NewTopic myClusterTopic6(){
//        return TopicBuilder
//                .name("my_cluster_topic6") //主题名称
//                .partitions(3) //分区数
//                .replicas(3) // 副本数
//                .build();
//
//    }
    //创建重试队列，以后kafka可以自动根据重试队列 加上dlt 后缀创建一个死信队列
//    @Bean
//    public NewTopic springTestRetryTopic(){
//        return TopicBuilder.name("spring_test_retry_topic") //主题名称
//                .partitions(3) //分区数量
//                .replicas(3) //副本数量
//                .build();
//    }
}
